package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.Arrays;

@Configuration
@Slf4j
public class KafkaStreamHelloListener {
    @Bean
    public KStream<String,String> kStream(StreamsBuilder builder){
        //获取数据，指定topic名字
        KStream<String, String> stream = builder.stream(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
        //计算方法写法和stream流处理基本一致
        //value "hello kafka"  -> [value : "hello",value : "kakfa"]
        stream.map((key, value) -> {
            //1.将value对象进行还原
            UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);

            return new KeyValue<>(updateArticleMess.getArticleId().toString(),value);
        }).groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //聚合
                .aggregate(() -> JSON.toJSONString(new ArticleVisitStreamMess()),
                        (key, value, aggregate) -> {
                            //aggregate就是上一次聚合之后的结果 ArticleVisitStreamMess-String
                            UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);
                            ArticleVisitStreamMess mess = JSON.parseObject(aggregate, ArticleVisitStreamMess.class);
                            //0.给文章ID赋值
                            if(mess.getArticleId() == null){
                                mess.setArticleId(Long.valueOf(key));
                            }

                            //1.解析出数据中的操作类型likes comment
                            switch (updateArticleMess.getType()){
                                case COLLECTION:
                                    //2.对应的域上，增加 增量 1  / - 1
                                    mess.setCollect(mess.getCollect() + updateArticleMess.getAdd());
                                    break;
                                case LIKES:
                                    mess.setLike(mess.getLike() + updateArticleMess.getAdd());
                                    break;
                                case VIEWS:
                                    mess.setView(mess.getView() + updateArticleMess.getAdd());
                                    break;
                                case COMMENT:
                                    mess.setComment(mess.getComment() + updateArticleMess.getAdd());
                                    break;
                                default:
                                    break;
                            }

                            //3.结果返回
                            return JSON.toJSONString(mess);
                        })

                .toStream()
                //将key value调整成String类型
                .map((key, value) -> new KeyValue<>(key.key(), value))
                //投递结果到kafka
                .to(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);


        return stream;
    }
}